package join.etc;


import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;


import java.io.IOException;
import java.util.ArrayList;

public class MapReduceJoin {
    public static class MapReduceJoinMapper extends Mapper<LongWritable, Text, Text, JoinBean> {
        String fileName = null;
        JoinBean bean = new JoinBean();
        Text text = new Text();


        @Override
        protected void map(LongWritable key, Text value, Mapper<LongWritable, Text, Text, JoinBean>.Context context) throws IOException, InterruptedException {
            String[] fields = value.toString().split(",");
            if (fileName.startsWith("order")) {
                bean.set(fields[0], fields[1], "NULL", -1, "NULL", "order");
            } else {
                bean.set("NULL", fields[0], fields[1], Integer.parseInt(fields[2]), fields[3], "user");
            }
            text.set(bean.getUserId());
            context.write(text, bean);
        }
    }

    public static class MapReduceJoinReducer extends Reducer<Text, JoinBean, JoinBean, NullWritable> implements join.etc.MapReduceJoinReducer {

        protected void Reduce(Text key, Iterable<JoinBean> beans, Reducer<Text, JoinBean, JoinBean, NullWritable>.Context context) throws IOException, InterruptedException {
            ArrayList<JoinBean> orderList = new ArrayList<JoinBean>();
            JoinBean userBean = null;
            try {
                for (JoinBean bean : beans) {
                    if ("order".equals(bean.getTableName())) {
                        JoinBean newBean = new JoinBean();
                        BeanUtils.copyProperties(newBean, bean);
                        orderList.add(newBean);
                    } else {
                        userBean = new JoinBean();
                        BeanUtils.copyProperties(userBean, bean);
                    }
                }
for (JoinBean bean :orderList){
    bean.setUserName(userBean.getUserName());
    bean.setUserAge(userBean.getUserAge());
    bean.setUserOuxiang(userBean.getUserOux());


    context.write(bean, NullWritable.get());


}
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws InterruptedException, IOException, ClassNotFoundException {
        Configuration conf = new Configuration();

        Job job = Job.getInstance(conf);
        job.setJarByClass(MapReduceJoin.class);

        job.setMapperClass(MapReduceJoinMapper.class);
        job.setReducerClass(MapReduceJoinReducer.class);

        job.setNumReduceTasks(2);

        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(JoinBean.class);

        job.setOutputKeyClass(JoinBean.class);
        job.setOutputValueClass(NullWritable.class);

        FileInputFormat.setInputPaths(job,new Path("F:\\join"));
        FileOutputFormat.setOutputPath(job,new Path("F:\\jt"));
        job.waitForCompletion(true);
    }

}